package flume_avro_client;

import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientFactory;
import org.apache.flume.event.EventBuilder;
import java.nio.charset.Charset;

public class AVRO_Client {
    public static void main(String[] args) {
        MyRpcClientFacade client = new MyRpcClientFacade();
        // Initializeclient with the remote Flume agent's host and port  
        //client.init("host.example.org",41414);  
        client.init("localhost", 42424);

        // Send 10events to the remote Flume agent.  
        // configured tolisten with an AvroSource.  
        String sampleData = "Hello Flume!";
        for (int i = 0; i < 10; i++) {
            client.sendDataToFlume(sampleData);
        }

        client.cleanUp();
    }
}

class MyRpcClientFacade {
    private RpcClient client;
    private String hostname;
    private int port;

    public void init(String hostname, int port) {
        // Setup the RPCconnection  
        this.hostname = hostname;
        this.port = port;
        this.client = RpcClientFactory.getDefaultInstance(hostname, port);
        // Use thefollowing method to create a thrift client (instead of the above line):  
        // this.client = RpcClientFactory.getThriftInstance(hostname, port);  
    }

    public void sendDataToFlume(String data) {
        // Create aFlume Event object that encapsulates the sample data  
        Event event = EventBuilder.withBody(data, Charset.forName("UTF-8"));

        // Send theevent  
        try {
            client.append(event);
        } catch (EventDeliveryException e) {
            // clean up andrecreate the client  
            client.close();
            client = null;
            client = RpcClientFactory.getDefaultInstance(hostname, port);
            // Use thefollowing method to create a thrift client (instead of the above line):  
            // this.client =RpcClientFactory.getThriftInstance(hostname, port);  
        }
    }

    public void cleanUp() {
        // Close the RPCconnection  
        client.close();
    }

}